package com.haoxin.stream.base

import com.haoxin.stream.util.ConfManager
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

trait FlinkStreamEnv {

  protected lazy val env: StreamExecutionEnvironment = initialize()

  private def initialize(): StreamExecutionEnvironment = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
//    env.enableCheckpointing(ConfManager.getLong("flink.checkpoint.interval.ms"),CheckpointingMode.AT_LEAST_ONCE)

    env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    env
  }

  protected def execute(args: Array[String])

  protected def run(args: Array[String]): Unit = {
    execute(args: Array[String])
  }


}
